package com.yang.simple_rpc.server;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;



/**
 * Netty RPC服务器，支持Netty网络服务开启关闭
 *
 * @author : morikawaaaa
 * @date : 2022
 */
public class NettyRpcServer extends RpcServer{

    private static Logger logger = LoggerFactory.getLogger(NettyRpcServer.class);

    private Channel channel;

    public NettyRpcServer(int port, String protocol, RequestHandler requestHandler){
        super(port, protocol, requestHandler);
    }

    @Override
    public void start() {
        //配置服务器
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workGroup = new NioEventLoopGroup();
        try{
            ServerBootstrap b = new ServerBootstrap();
            /**设置服务端最大连接数100，分别给workgroup和bossgroup进行配置，只需要给workgroup的
            handler里面设置pipeline，添加ChannelRequestHandler
            */b.group(bossGroup, workGroup).channel(NioServerSocketChannel.class).option(ChannelOption.SO_BACKLOG, 100)
                    .handler(new LoggingHandler(LogLevel.INFO)).childHandler(new ChannelInitializer<SocketChannel>() {
                @Override
                protected void initChannel(SocketChannel socketChannel){
                    ChannelPipeline cp = socketChannel.pipeline();
                    cp.addLast(new ChannelRequestHandler());
                }
            });

            //启动服务
            ChannelFuture f = b.bind(port).sync();
            logger.info("Server started successfully.");
            channel = f.channel();
            f.channel().closeFuture().sync();

        }catch (Exception e){
            e.printStackTrace();
        }finally {
            bossGroup.shutdownGracefully();
            workGroup.shutdownGracefully();
        }
    }


    @Override
    public void stop(){
        try{
            this.channel.close();
        }catch (Exception e){
            e.printStackTrace();
        }
    }


    private class ChannelRequestHandler extends ChannelInboundHandlerAdapter{

        @Override
        public void channelActive(ChannelHandlerContext ctx) {
            logger.info("Channel active : {}", ctx);
        }

        @Override
        public void channelRead(ChannelHandlerContext ctx, Object obj) throws Exception {
            logger.info("The server receive a message {}", obj);
            //从bytebuf读出
            ByteBuf reqbuf = (ByteBuf) obj;
            byte[] req = new byte[reqbuf.readableBytes()];
            reqbuf.readBytes(req);

            byte[] res = handler.handleRequest(req);
            logger.info("Send message : {}", obj);
            //写入发送缓存
            ByteBuf respBuf = Unpooled.buffer(res.length);
            respBuf.writeBytes(res);
            ctx.write(respBuf);

        }

        @Override
        public void channelReadComplete(ChannelHandlerContext ctx){
            ctx.flush();
        }

        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)  {
            cause.printStackTrace();
            logger.error("Exception occured : {}", cause.getMessage());
            ctx.close();
        }
    }
}
